package com.example.flinkcourse.lesson2;

import com.example.flinkcourse.lesson2.model.Event;
import com.example.flinkcourse.lesson2.source.PulsarSourceFactory;
import com.example.flinkcourse.lesson2.stream.EventProcessor;
import com.example.flinkcourse.lesson2.table.EventTableProcessor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Flink Lesson 2 主作业类
 * 演示 DataStream API 和 Table API 的使用
 */
public class FlinkLesson2Job {
    private static final Logger LOG = LoggerFactory.getLogger(FlinkLesson2Job.class);

    public static void main(String[] args) throws Exception {
        LOG.info("Starting Flink Lesson 2 Job...");
        
        // 创建执行环境
        Configuration configuration = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        
        // 配置检查点
        configureCheckpoint(env);
        
        // 创建 Pulsar 数据源
        DataStream<Event> inputStream = env.fromSource(
            PulsarSourceFactory.createPulsarSource(),
            WatermarkStrategy.noWatermarks(),
            "Pulsar Source"
        ).setParallelism(2);
        
        // DataStream API 处理
        DataStream<Event> processedStream = EventProcessor.processEvents(inputStream);
        
        // Table API 处理
        Table resultTable = EventTableProcessor.processEvents(env, processedStream);
        
        // 输出结果
        resultTable.executeInsert("output_table");
        
        // 执行作业
        LOG.info("Executing Flink Lesson 2 Job...");
        env.execute("Flink Lesson 2 Job");
    }
    
    /**
     * 配置检查点
     * @param env 流执行环境
     */
    private static void configureCheckpoint(StreamExecutionEnvironment env) {
        LOG.info("Configuring checkpoint...");
        
        // 启用检查点
        env.enableCheckpointing(60000);
        
        // 设置检查点模式
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        
        // 设置检查点超时
        env.getCheckpointConfig().setCheckpointTimeout(30000);
        
        // 设置检查点之间的最小暂停时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);
        
        // 设置最大并发检查点数量
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        
        LOG.info("Checkpoint configuration completed");
    }
} 